package com.millstein.realtime.app.dwd.db;

import com.millstein.realtime.app.base.BaseSqlApp;
import com.millstein.realtime.common.Constants;
import com.millstein.realtime.util.SqlUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * @Description
 * @Author tsing
 * @Date 2024-10-14 16:09
 */
public class Dwd_15_InteractionComment extends BaseSqlApp {

    public static void main(String[] args) {
        new Dwd_15_InteractionComment().init(
                7009,
                3,
                "Dwd_15_InteractionComment"
        );
    }

    /**
     * 具体数据处理的逻辑，由子类编写
     *
     * @param env      执行环境
     * @param tableEnv 表执行环境
     */
    @Override
    public void handle(StreamExecutionEnvironment env, StreamTableEnvironment tableEnv) {
        // 1.从ods层中读取数据
        readOdsDataFromKafka(tableEnv, "Dwd_15_InteractionComment");

        // 2.过滤出评论数据
        Table commentInfo = tableEnv.sqlQuery(
                "select " +
                "    `data`['id'] id, " +
                "    `data`['user_id'] user_id, " +
                "    `data`['sku_id'] sku_id, " +
                "    `data`['order_id'] order_id, " +
                "    `data`['create_time'] create_time, " +
                "    `data`['appraise'] appraise, " +
                "    `pt`, " +
                "    `ts` " +
                "from maxwell_table " +
                "where `database` = 'gmall' " +
                "    and `table` = 'comment_info' " +
                "    and `type` = 'insert'"
        );
        tableEnv.createTemporaryView("comment_info", commentInfo);

        // 3.读取字典表中的数据
        readBaseDicFromMysql(tableEnv);

        // 4.进行维度退化
        Table resultTable = tableEnv.sqlQuery(
                "select " +
                "    ci.id, " +
                "    ci.user_id, " +
                "    ci.sku_id, " +
                "    ci.order_id, " +
                "    date_format(ci.create_time, 'yyyy-MM-dd') date_id, " +
                "    ci.create_time, " +
                "    ci.appraise appraise_code, " +
                "    bd.dic_name appraise_name, " +
                "    ci.ts " +
                "from comment_info ci " +
                "join base_dic for system_time as of ci.pt as bd on ci.appraise = bd.dic_code"
        );

        // 5.创建kafka-sink的动态表
        tableEnv.executeSql(
                "create table dwd_interaction_comment ( " +
                "    id string, " +
                "    user_id string, " +
                "    sku_id string, " +
                "    order_id string, " +
                "    date_id string, " +
                "    create_time string, " +
                "    appraise_code string, " +
                "    appraise_name string, " +
                "    ts bigint " +
                ")" + SqlUtil.getKafkaSinkDDL(Constants.TOPIC_DWD_INTERACTION_COMMENT)
        );

        // 6.将最终的数据写入动态表中
        resultTable.executeInsert("dwd_interaction_comment");
    }
}
